package io.github.wanggit.antrpc.client.zk.register;

import io.github.wanggit.antrpc.client.zk.zknode.IZkNodeOperator;
import io.github.wanggit.antrpc.commons.annotations.OnRpcFail;
import io.github.wanggit.antrpc.commons.annotations.RpcMethod;
import io.github.wanggit.antrpc.commons.annotations.RpcService;
import io.github.wanggit.antrpc.commons.bean.IpNodeDataBean;
import io.github.wanggit.antrpc.commons.bean.RegisterBean;
import io.github.wanggit.antrpc.commons.bean.RegisterBeanHelper;
import io.github.wanggit.antrpc.commons.bean.RegisterBeanMethod;
import io.github.wanggit.antrpc.commons.config.IConfiguration;
import io.github.wanggit.antrpc.commons.constants.ConstantValues;
import io.github.wanggit.antrpc.commons.constants.Constants;
import io.github.wanggit.antrpc.commons.utils.AntRpcClassUtils;
import io.github.wanggit.antrpc.commons.utils.ApplicationNameUtil;
import io.github.wanggit.antrpc.commons.utils.JsonUtils;
import io.github.wanggit.antrpc.commons.zookeeper.NodeUnregDTO;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.reflect.MethodUtils;
import org.apache.zookeeper.CreateMode;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.core.annotation.AnnotationUtils;

import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;

/** 获取所有标记了{@link RpcService}注解的对象，并把这些对象注册为RPC服务。 */
@Slf4j
public class ZkRegister implements IRegister {

    private final List<RegisterBean> registerBeans = new ArrayList<>();

    private IConfiguration configuration;

    /** */
    @Override
    public void register(
            RegisterBean registerBean, IZkNodeOperator zkNodeOperator, String exposeIp) {
        String zkFullpath = registerBean.getZookeeperFullPath(exposeIp);
        // 判断父节点是否存在，如果父节点不存在就创建父节点。因为父节点上带数据，所以不能直接创建父节点
        String parentFullPath = registerBean.getZookeeperParentFullPath(exposeIp);
        if (!zkNodeOperator.existsNode(parentFullPath)) {
            createIpNode(zkNodeOperator, configuration, parentFullPath);
        }
        if (!zkNodeOperator.existsData(parentFullPath)) {
            setIpNodeData(zkNodeOperator, configuration, parentFullPath);
        }

        // 判断当前节点是否已被下线
        String unregFullpath =
                "/"
                        + ConstantValues.ZK_ROOT_NODE_UNREG_NAME
                        + "/"
                        + exposeIp
                        + ":"
                        + registerBean.getPort()
                        + "/"
                        + registerBean.getClassName();
        boolean wasOfflined = zkNodeOperator.existsNode(unregFullpath);
        if (wasOfflined && log.isInfoEnabled()) {
            log.info(unregFullpath + " was existed. the interface offlined. can not register.");
        }
        if (!wasOfflined) {
            byte[] nodeData =
                    registerBean.getNodeData(
                            configuration.getApplicationName(),
                            zkFullpath,
                            registerBean.getClassName(),
                            exposeIp + ":" + registerBean.getPort());
            zkNodeOperator.remoteCreateZkNode(zkFullpath, nodeData, CreateMode.EPHEMERAL);
        }
    }

    @Override
    public void unregister(IConfiguration configuration, IZkNodeOperator zkNodeOperator) {
        if (log.isInfoEnabled()) {
            log.info("All service will be unregister.");
        }
        for (RegisterBean registerBean : registerBeans) {
            String zookeeperFullPath =
                    registerBean.getZookeeperFullPath(configuration.getExposeIp());
            if (log.isInfoEnabled()) {
                log.info("The " + zookeeperFullPath + " node will be deleted.");
            }
            zkNodeOperator.deleteNode(zookeeperFullPath);
        }
    }

    @Override
    public List<RegisterBean> snapshot() {
        return new ArrayList<>(registerBeans);
    }

    @Override
    public void playAllRegister(IConfiguration configuration, IZkNodeOperator zkNodeOperator) {
        registerBeans.forEach(
                it -> {
                    it.setPause(false);
                    deleteNodeUnregPath(it, configuration, zkNodeOperator);
                });
    }

    @Override
    public void playRegister(
            String className, IConfiguration configuration, IZkNodeOperator zkNodeOperator) {
        RegisterBean registerBean = findRegisterBeanByClassName(className);
        if (null != registerBean) {
            registerBean.setPause(false);
            deleteNodeUnregPath(registerBean, configuration, zkNodeOperator);
        }
    }

    @Override
    public void pauseAllRegister(IConfiguration configuration, IZkNodeOperator zkNodeOperator) {
        registerBeans.forEach(
                it -> {
                    it.setPause(true);
                    zkNodeOperator.deleteNode(it.getZookeeperFullPath(configuration.getExposeIp()));
                    createNodeUnregPath(it, configuration, zkNodeOperator);
                });
    }

    @Override
    public void pauseRegister(
            IConfiguration configuration, IZkNodeOperator zkNodeOperator, String className) {
        RegisterBean registerBean = findRegisterBeanByClassName(className);
        if (null != registerBean) {
            registerBean.setPause(true);
            zkNodeOperator.deleteNode(
                    registerBean.getZookeeperFullPath(configuration.getExposeIp()));
            createNodeUnregPath(registerBean, configuration, zkNodeOperator);
        }
    }

    @Override
    public RegisterBean findRegisterBeanByClassName(String className) {
        for (RegisterBean bean : registerBeans) {
            if (bean.getClassName().equals(className)) {
                return bean;
            }
        }
        return null;
    }

    // 1
    @Override
    public void checkHasRpcService(Object bean) {
        Class<?> beanClass = bean.getClass();
        RpcService rpcService = AnnotationUtils.findAnnotation(beanClass, RpcService.class);
        OnRpcFail onRpcFail = AnnotationUtils.findAnnotation(beanClass, OnRpcFail.class);
        // 同时被标记@RpcService与@OnRpcFail注解的对象不能被发布为RPC服务
        if (null != rpcService && null != onRpcFail) {
            if (log.isWarnEnabled()) {
                log.warn(
                        "The "
                                + beanClass.getName()
                                + " marked with "
                                + OnRpcFail.class.getSimpleName()
                                + " annotations cannot be published as RPC services");
            }
        }
        // 标识OnRpcFail注解的对象不能被注册到注册中心，只能在本地使用
        if (null != rpcService && null == onRpcFail) {
            internalCreateRegisterBean(rpcService, bean, beanClass);
        }
    }

    private void internalCreateRegisterBean(
            RpcService rpcService, Object bean, Class<?> beanClass) {
        String className = beanClass.getName();
        List<Class<?>> classes = internalGetInterfaces(rpcService, bean);
        for (Class<?> interfaceClass : classes) {
            String interfaceClassName = interfaceClass.getName();
            if (log.isInfoEnabled()) {
                log.info(
                        className
                                + " will be registered to zookeeper as "
                                + interfaceClassName
                                + " interface type.");
            }
            Method[] methods =
                    MethodUtils.getMethodsWithAnnotation(beanClass, RpcMethod.class, true, false);
            RegisterBean registerBean = new RegisterBean();
            registerBean.setClassName(interfaceClassName);
            // 如果对象上标记了 @RpcService 注解，但是没有对象里没有任何一个方法标记 @RpcMethod 那么也报错
            if (methods.length == 0) {
                throw new BeanCreationException(
                        beanClass.getName() + " Not Found any method has @RpcMethod.");
            }
            for (Method method : methods) {
                RegisterBeanMethod registerBeanMethod =
                        RegisterBeanHelper.getRegisterBeanMethod(method);
                registerBean.addMethod(registerBeanMethod);
            }
            registerBeans.add(registerBean);
        }
    }

    private List<Class<?>> internalGetInterfaces(RpcService rpcService, Object bean) {
        List<Class<?>> classes = new ArrayList<>();
        // 如果RpcService注册中未指定接口，那么就此类的第一个接口
        if (rpcService.clazzes().length == 0) {
            List<Class<?>> allInterfaces = AntRpcClassUtils.getAllInterfaces(bean);
            if (allInterfaces.isEmpty()) {
                throw new BeanCreationException(
                        "The @"
                                + RpcService.class.getSimpleName()
                                + " tagged class must implement an interface.");
            }
            Class<?> interfaceClass = allInterfaces.get(0);
            // 超过一个接口，而又未指定接口
            if (allInterfaces.size() > 1) {
                if (log.isWarnEnabled()) {
                    log.warn(
                            bean.getClass().getName()
                                    + " implements many interfaces, will choose the "
                                    + interfaceClass.getName());
                }
            }
            classes.add(interfaceClass);
        } else {
            for (Class<?> interfaceClass : rpcService.clazzes()) {
                // 有正确的接口实现关系，才可被使用
                if (interfaceClass.isAssignableFrom(bean.getClass())) {
                    classes.add(interfaceClass);
                } else {
                    throw new BeanCreationException(
                            bean.getClass().getName()
                                    + " Not Implements "
                                    + interfaceClass.getName());
                }
            }
        }
        return classes;
    }

    // 3
    @Override
    public void init(
            IZkNodeOperator zkNodeOperator,
            IZkRegisterHolder zkRegisterHolder,
            IConfiguration configuration) {
        this.configuration = configuration;
        Integer rpcPort = configuration.getPort();
        if (null == rpcPort) {
            if (log.isWarnEnabled()) {
                log.warn(
                        "The configured port cannot be found. Check that "
                                + Constants.RPC_PORT_PROP_NAME
                                + " is configured.");
            }
        } else {
            if (log.isInfoEnabled()) {
                log.info("Get to the configuration port is " + rpcPort);
            }
        }
        String fullPath =
                "/"
                        + ConstantValues.ZK_ROOT_NODE_NAME
                        + "/"
                        + configuration.getExposeIp()
                        + (null == rpcPort ? "" : ":" + rpcPort);
        createIpNode(zkNodeOperator, configuration, fullPath);

        for (RegisterBean registerBean : registerBeans) {
            registerBean.setPort(rpcPort);
            zkRegisterHolder.add(registerBean);
            register(registerBean, zkNodeOperator, configuration.getExposeIp());
        }
    }

    private void setIpNodeData(
            IZkNodeOperator zkNodeOperator, IConfiguration configuration, String fullPath) {
        byte[] nodeData = createIpNodeDataBytes(configuration);
        zkNodeOperator.setNodeData(fullPath, nodeData);
    }

    private void createIpNode(
            IZkNodeOperator zkNodeOperator, IConfiguration configuration, String fullPath) {
        byte[] nodeData = createIpNodeDataBytes(configuration);
        zkNodeOperator.remoteCreateZkNode(fullPath, nodeData, CreateMode.PERSISTENT);
    }

    private byte[] createIpNodeDataBytes(IConfiguration configuration) {
        IpNodeDataBean ipNodeDataBean = new IpNodeDataBean();
        ipNodeDataBean.setAppName(
                ApplicationNameUtil.getApplicationName(
                        configuration.getExposeIp(),
                        configuration.getApplicationName(),
                        configuration.getPort()));
        ipNodeDataBean.setTs(System.currentTimeMillis());
        ipNodeDataBean.setRpcPort(configuration.getPort());
        ipNodeDataBean.setHttpPort(
                configuration.getEnvironment().getProperty("server.port", Integer.class));
        return JsonUtils.toJsonBytes(ipNodeDataBean);
    }

    private void deleteNodeUnregPath(
            RegisterBean registerBean,
            IConfiguration configuration,
            IZkNodeOperator zkNodeOperator) {
        String host = configuration.getExposeIp() + ":" + configuration.getPort();
        zkNodeOperator.deleteNode(
                "/"
                        + ConstantValues.ZK_ROOT_NODE_UNREG_NAME
                        + "/"
                        + host
                        + "/"
                        + registerBean.getClassName());
    }

    private void createNodeUnregPath(
            RegisterBean registerBean,
            IConfiguration configuration,
            IZkNodeOperator zkNodeOperator) {
        NodeUnregDTO nodeUnregDTO = new NodeUnregDTO();
        nodeUnregDTO.setClazz(registerBean.getClassName());
        String host = configuration.getExposeIp() + ":" + configuration.getPort();
        nodeUnregDTO.setHost(host);
        nodeUnregDTO.setAppName(configuration.getApplicationName());
        nodeUnregDTO.setTs(System.currentTimeMillis());
        zkNodeOperator.remoteCreateZkNode(
                "/"
                        + ConstantValues.ZK_ROOT_NODE_UNREG_NAME
                        + "/"
                        + host
                        + "/"
                        + registerBean.getClassName(),
                JsonUtils.toJsonBytes(nodeUnregDTO),
                CreateMode.EPHEMERAL);
    }
}
